package com.panlm.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;

public class TimeInterceptor implements ProducerInterceptor<String, String> {

    private DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss");

    /**
     * 获取配置信息和初始化数据时调用
     */
    @Override
    public void configure(Map<String, ?> map) {

    }

    /**
     * 在消息被序列化以及计算分区前调用该方法
     * 用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响目标分区的计算
     * @param producerRecord
     * @return
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        //创建一个新的record，把时间戳写入消息体的最前部
        ProducerRecord<String, String> record = new ProducerRecord<>(producerRecord.topic(), producerRecord.partition(),
                producerRecord.key(), formatter.format(LocalDateTime.now()) +"-"+producerRecord.value());
        return record;
    }

    /**
     * 该方法会在消息从RecordAccumulator成功发送到Kafka Broker之后，或者在发送过程中失败时调用
     * 并且通常都是在producer回调逻辑触发之前。onAcknowledgement 运行在producer的IO线程中，
     * 因此不要在该方法中放入很重的逻辑，否则会拖慢producer的消息发送效率。
     * @param recordMetadata
     * @param e
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    /**
     * 关闭interceptor，主要用于执行一些资源清理工作
     */
    @Override
    public void close() {

    }


}
